package register;

import Config.RegistryConfig;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;
import model.ServiceInfo;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * EtcdRegistry：RPC 框架默认的注册中心实现，基于 Etcd
 */
@Slf4j
public class EtcdRegistry implements Registry {
    /** etcd 客户端 */
    private Client client;
    /** etcd 键值客户端 */
    private KV kvClient;
    /** etcd 服务注册根路径 */
    private static final String DEFAULT_ETCD_ROOT_PATH = "/rpc/";

    /** 
     * 本地已注册服务节点的键集合（如某个 ServiceInfo 的 key），用于心跳检测和服务下线 
     */
    private Set<String> localRegistryServiceNodeSet = new HashSet<>();

    /** 
     * 服务本地缓存（目前仅支持单服务） 
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();

    /**
     * 正在监听的 key 集合，防止重复监听
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();

    /**
     * 注册中心初始化（连接注册中心，服务消费者和服务提供者都要初始化）
     * 
     * @param registryConfig 注册中心配置
     */
    public void init(RegistryConfig registryConfig) {
        client = Client.builder()
                .endpoints(registryConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        // 启动心跳检测
        heartBeat();
        log.info("注册中心开始心跳检测......");
    }

    /**
     * 注册服务元信息到注册中心（服务提供者调用）
     * 
     * @param serviceInfo 服务元信息
     * @throws Exception 注册异常
     */
    public void register(ServiceInfo serviceInfo) throws Exception {
        Lease leaseClient = client.getLeaseClient();
        // 默认 60 秒后删除键值对
        long leaseId = leaseClient.grant(30 * 2).get().getID();

        // 生成服务节点的 etcd 键，格式：/rpc/serviceName:version/serviceHost:servicePort
        String serviceKey = DEFAULT_ETCD_ROOT_PATH + serviceInfo.getServiceNodeKey();

        // 将当前注册节点键加入本地集合，用于心跳检测
        localRegistryServiceNodeSet.add(serviceKey);

        ByteSequence key = ByteSequence.from(serviceKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceInfo), StandardCharsets.UTF_8);
        // 绑定租约
        PutOption putRelationShip = PutOption.builder()
                .withLeaseId(leaseId)
                .build();
        kvClient.put(key, value, putRelationShip);
    }

    /**
     * 注销服务（服务提供者调用）
     * 
     * @param serviceInfo 服务元信息
     */
    public void unRegister(ServiceInfo serviceInfo) {
        // 生成要删除的服务键名
        String serviceKey = DEFAULT_ETCD_ROOT_PATH + serviceInfo.getServiceNodeKey();
        ByteSequence deleteKey = ByteSequence.from(serviceKey, StandardCharsets.UTF_8);
        kvClient.delete(deleteKey);
        // 从本地集合移除该服务键名
        localRegistryServiceNodeSet.remove(serviceKey);
    }

    /**
     * 服务发现（服务消费者调用）
     * 
     * @param serviceKey 服务唯一标识（serviceName:version）
     * @return 服务元信息列表
     */
    public List<ServiceInfo> serviceDiscovery(String serviceKey) {
        // 优先从本地缓存获取
        List<ServiceInfo> serviceInfoListCache = registryServiceCache.readCache();
        if (!CollUtil.isEmpty(serviceInfoListCache))
            return serviceInfoListCache;
        // 缓存未命中，从 etcd 查询
        String searchPrefix = DEFAULT_ETCD_ROOT_PATH + serviceKey + "/";
        try {
            // 前缀查询
            GetOption getOption = GetOption.builder()
                    .isPrefix(true)
                    .build();
            List<KeyValue> kvList = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), getOption)
                    .get()
                    .getKvs();
            // 查询结果转换为 ServiceInfo 列表，并对每个 key 添加监听
            List<ServiceInfo> serviceInfoList = kvList.stream().map(KV -> {
                String key = KV.getKey().toString(StandardCharsets.UTF_8);
                // 对该 key 实施监听，便于变更时触发缓存更新
                watchKey(key);
                String jsonValue = KV.getValue().toString(StandardCharsets.UTF_8);
                log.info("jsonValue:::::::" + jsonValue);
                return JSONUtil.toBean(jsonValue, ServiceInfo.class);
            }).collect(Collectors.toList());
            // 更新本地缓存
            registryServiceCache.writeCache(serviceInfoList);
            return serviceInfoList;
        } catch (Exception e) {
            throw new RuntimeException("获取:" + serviceKey + "服务列表失败");
        }
    }

    /**
     * 注销注册中心（项目关闭时调用）
     */
    public void destory() {
        // 释放客户端资源
        log.info("节点下线");
        log.info("" + localRegistryServiceNodeSet.size());
        // 下线所有节点（配合主动下线方案）
        for (String key : localRegistryServiceNodeSet) {
            try {
                log.info("下线的节点是：" + key);
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(key + "节点下线异常");
            }
        }
        if (!Objects.isNull(kvClient))
            kvClient.close();
        if (!Objects.isNull(client))
            client.close();
    }

    /**
     * 心跳检测（本地维护已注册服务节点，定期遍历节点，检查是否还在 etcd 注册中心，若有则续期）
     */
    public void heartBeat() {
        // 启动定时任务，每 30 秒扫描一次 localRegistryServiceNodeSet
        CronUtil.schedule("*/30 * * * * *", new Task() {
            public void execute() {
                for (String key : localRegistryServiceNodeSet) {
                    try {
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get()
                                .getKvs();
                        // 节点已过期
                        if (CollUtil.isEmpty(keyValues)) {
                            // 记录节点过期日志
                            log.info(key + "节点已过期");
                            continue;
                        }
                            
                        // 节点未过期，进行续期
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8); // 获取 value 的 json 字符串
                        ServiceInfo serviceInfo = JSONUtil.toBean(value, ServiceInfo.class);
                        register(serviceInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + "续签失败");
                    }
                }
            }
        });
        // 启用秒级调度
        CronUtil.setMatchSecond(true);
        // 启动定时任务线程
        CronUtil.start();
    }

    /**
     * 监听指定服务节点 key 的变更（如删除、更新），用于自动更新本地缓存
     * 
     * @param serviceNodeKey 服务节点的 etcd key
     */
    @Override
    public void watchKey(String serviceNodeKey) {
        // 获取 etcd 监听客户端
        Watch watchClient = client.getWatchClient();
        if (watchingKeySet.add(serviceNodeKey)) {
            // 当前 serviceNodeKey 未被监听，添加监听
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        case DELETE:
                            // 监听到 key 被删除，清空本地缓存（因为之前对 key 的修改都是删除操作）
                            registryServiceCache.deleteCache();
                            break;
                        case PUT:
                        default:
                            break;
                    }
                }
            });
        }
    }
}
